#include "defs.h"

#include "thread_pool.h"
#include "thread.h"
int g_thread_count = 0;
int s_stop = false;
using namespace std;

ThreadPool::ThreadPool(int max_thread) :
		_max_thread_count(max_thread),
		_cur_thread_count(0)
{
}

ThreadPool::~ThreadPool(){
	vector<Thread*>::iterator it;

	for (it =  _thread_list.begin();
		 it != _thread_list.end();
		 it ++ ) 
	{
		_queue.enqueue(EXIT_SIGN);	
	}
	for (it =  _thread_list.begin();
		 it != _thread_list.end();
		 it ++ ) 
	{
		Thread* thread = *it;
		FREE_POINTER(thread);
	}
}

void
ThreadPool::cont(){
	if (!s_stop)
		return;
	vector<Thread*>::iterator it;
	for (it =  _thread_list.begin();
		 it != _thread_list.end();
		 it ++ ) 
	{
		_queue.enqueue(CONTINUE_SIGN);	
	}
}
void
ThreadPool::start(){
	for (int i = 0; i < _max_thread_count; i++) {
		_add_a_thread();
	}
}

void
ThreadPool::_add_a_thread(){
	Thread* thread = new Thread(this);
	char name[20];
	snprintf(name, 20, "Thread (%d)", _cur_thread_count);
	thread->set_name(name);
	thread->start();
	_thread_list.push_back(thread);
	AtomicAdd(_cur_thread_count, 1);
}
void
ThreadPool::stop(){
	if(s_stop)
		return;
	vector<Thread*>::iterator it;

	for ( it =  _thread_list.begin();
		  it != _thread_list.end();
		  it ++)
	{
		_queue.enqueue(STOP_SIGN);
	}
	s_stop = true;
}

bool
ThreadPool::queue_job(Job* job) {
	if (s_stop) {
		return false;
	}
	TRACE_LOG("enqueue a job (%s)", job->name.c_str());
	job->add_ref();
	_queue.enqueue(job);
	return true;
}

Job*
ThreadPool::get_job(long timeout){
	if(timeout == 0) {
		return _queue.dequeue();
	}
	else
		return _queue.dequeue(timeout);
}
